package com.zookeeper.MasterWorker;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.Random;

/**
 * Created by peak on 2015/11/30.
 */
public class Master  implements Watcher {
    private ZooKeeper zk;

    private String connnectHost;

    public Master(String connnectHost){
        this.connnectHost = connnectHost;
    }

    public void startZK() throws IOException {
        zk = new ZooKeeper(connnectHost, 15000, this);
    }

    public void process(WatchedEvent event) {
        System.out.print(event);
    }

    Random random = new Random();
    // ��ʶ�Լ�
    private String serverId = Integer.toHexString(random.nextInt());
    // �Ƿ���Leader
    private boolean isLeader;

    public void runForMaster() throws KeeperException, InterruptedException {
        while (true) {
            try {
                zk.create("/master", serverId.getBytes(),
                        ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
                // ����/master�ڵ�ɹ�
                isLeader = true;
                break;
            } catch (KeeperException.NodeExistsException e) {
                // /master�Ѿ�����
                isLeader = false;
                break;
            } catch (KeeperException.ConnectionLossException e) {
            }
            if (checkMaster()) break;
        }
    }

    /**
     * ���master�Ƿ����
     * @return ���ڷ���true����֮false
     */
    private boolean checkMaster() throws InterruptedException{
        while (true) {
            try {
                Stat stat = new Stat();
                byte data[] = zk.getData("/master", false, stat);
                isLeader = new String(data).equals(serverId);
                return true;
            } catch (KeeperException.NoNodeException e) {
                // û��master�ڵ㣬�ɷ���false
                return false;
            } catch (KeeperException e) {
                e.printStackTrace();
            }
        }
    }
    private AsyncCallback.StringCallback masterCreateCb = new AsyncCallback.StringCallback() {
        public void processResult(int rc, String path, Object ctx, String name) {
            switch (KeeperException.Code.get(rc)){
                case CONNECTIONLOSS:
//                    checkMaster();
                    return;
                case OK: // �����ڵ�ɹ����ɹ���ȡ�쵼Ȩ
                    isLeader = true;
                    break;
                default:
                    isLeader = false;
            }
            System.out.println("I'm " + (isLeader ? "" : "not ") + "the leader");
        }
    };

    private AsyncCallback.DataCallback masterCheckCb = new AsyncCallback.DataCallback() {
        public void processResult(int rc, String path, Object ctx, byte[] data, Stat stat) {
            switch(KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
//                    checkMaster();
                    return;
                case NONODE: // û��master�ڵ���ڣ����Ի�ȡ�쵼Ȩ
//                    runForMaster();
                    return;
                case NODEEXISTS:
                    System.out.println("node exists.");
            }
        }
    };

    /*public void bootstrap() {
        // ��ſ��õ�Worker
        createParent("/workers", new byte[0]);
        // ���Worker�´�ŵ�����
        createParent("/assign", new byte[0]);
        // ��ִ�к����ύ������
        createParent("/tasks", new byte[0]);
        createParent("/status", new byte[0]);
    }*/

    /*private void createParent(String path, byte[] data) {
        zk.create(path,
                data,
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.PERSISTENT,
                createParentCb,
                data);

    private AsyncCallback.StringCallback createParentCb  = new AsyncCallback.StringCallback() {
        public void processResult(int rc, String path, Object ctx, String name) {
            switch (KeeperException.Code.get(rc)) {
                case CONNECTIONLOSS:
                    createParent(path, (byte[]) ctx); //ȡ��data
                    break;
                case OK:
                    System.out.println("Parent created");
                    break;
                case NODEEXISTS:
                    System.out.println("Parent already registered: " + path);
                    break;
                default:
                    System.err.println("Something went wrong: " + KeeperException.create(KeeperException.Code.get(rc), path));
            }
        }
    };*/

    public static void main(String[] args) throws Exception {
        String connnectHost = "192.168.112.130:2182,192.168.112.130:2183,192.168.112.130:2184";
        Master m = new Master(connnectHost);
        m.startZK();

        m.runForMaster();
        if (m.isLeader){
            System.out.println("I am leader");
        } else {
            System.out.println("I am not leader");
        }

        Thread.sleep(3660000);
    }
}
